package MapReduce.TopN;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TopNReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private TreeMap<WordFrequencyPair, Integer> topWords;
    private Text outK = new Text();
    private IntWritable outV = new IntWritable();

    @Override
    protected void setup(Context context) {
        topWords = new TreeMap<>();
    }

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) {
        int frequency = 0;
        for (IntWritable value : values) {
            frequency += value.get();
        }
        //top50
        //这样，即使在词频相同的情况下，通过词语作为区分标识，也能正确保存所有词频前50的词语及其词频。
        WordFrequencyPair pair = new WordFrequencyPair(frequency, key.toString());
        topWords.put(pair, frequency);
        if (topWords.size() > 50) {
            topWords.remove(topWords.firstKey());
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {

        // 此处做了倒序处理 --将递增序列变为递减序列 descendingMap()
        for (Map.Entry<WordFrequencyPair, Integer> entry : topWords.descendingMap().entrySet()) {

            // 从treeMap中取值
            WordFrequencyPair pair = entry.getKey();
            int frequency = entry.getValue();

            // 此处封装
            outK.set(pair.getWord());
            outV.set(frequency);

            // 此处写出
            context.write(outK, outV);
        }
    }
}
